package cn._51doit.live.udfs;

import cn._51doit.live.pojo.DataBean;
import cn._51doit.live.pojo.GiftBean;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;

public class GiftCountFunction extends KeyedBroadcastProcessFunction<String, DataBean, GiftBean, Tuple3<String, Integer, Double>> {

    private MapStateDescriptor<Integer, Tuple2<String, Double>> stateDescriptor;

    private transient ValueState<Integer> giftCountState;
    private transient ValueState<Double> giftPointState;

    public GiftCountFunction(MapStateDescriptor<Integer, Tuple2<String, Double>> stateDescriptor) {
        this.stateDescriptor = stateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //初始化状态
        ValueStateDescriptor<Integer> stateDescriptor1 = new ValueStateDescriptor<>("gift-count-state", Integer.class);
        giftCountState = getRuntimeContext().getState(stateDescriptor1);

        ValueStateDescriptor<Double> stateDescriptor2 = new ValueStateDescriptor<>("gift-count-state", Double.class);
        giftPointState = getRuntimeContext().getState(stateDescriptor2);
    }

    @Override
    public void processElement(DataBean value, ReadOnlyContext ctx, Collector<Tuple3<String, Integer, Double>> out) throws Exception {
        String anchorId = ctx.getCurrentKey();
        String giftId = value.getProperties().get("gift_id").toString();

        ReadOnlyBroadcastState<Integer, Tuple2<String, Double>> broadcastState = ctx.getBroadcastState(stateDescriptor);
        Tuple2<String, Double> tp = broadcastState.get(Integer.parseInt(giftId));
        Double points = tp.f1;

        Integer totalCount = giftCountState.value();
        if (totalCount == null) {
            totalCount = 0;
        }
        totalCount += 1;
        giftCountState.update(totalCount);

        Double totalPoints = giftPointState.value();
        if (totalPoints == null) {
            totalPoints = 0.0;
        }
        totalPoints += points;

        out.collect(Tuple3.of(anchorId, totalCount, totalPoints));

    }

    @Override
    public void processBroadcastElement(GiftBean value, Context ctx, Collector<Tuple3<String, Integer, Double>> out) throws Exception {

        Integer id = value.id;
        String name = value.name;
        Double point = value.point;
        int deleted = value.deleted;
        BroadcastState<Integer, Tuple2<String, Double>> broadcastState = ctx.getBroadcastState(stateDescriptor);
        if (deleted != 1) {
            //新增或更新
            broadcastState.put(id, Tuple2.of(name, point));
        } else {
            broadcastState.remove(id);
        }


    }
}
